package com.atguigu1.core.wc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 *
 * @description: 行动算子案例
 * @time: 2021-03-12 11:45
 * @author: baojinlong
 **/
object Spark04WordCountSummary {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    // 设置rdd分区数字
    val sparkContext = new SparkContext(conf)
    val inputRddValue: RDD[String] = sparkContext.makeRDD(Seq("Hello ajax", "Hello Scala", "Hello Spark", "Hello windy", "windy alice"), numSlices = 2)
    wordCount09(inputRddValue)
    sparkContext.stop()
  }

  /**
   * groupBy
   *
   * @param inputRddValue
   */
  def wordCount01(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val groupResult: RDD[(String, Iterable[String])] = rdd.groupBy(word => word)
    val result: RDD[(String, Int)] = groupResult.mapValues(item => item.size)
    result.collect.foreach(println)
  }

  /**
   * groupByKey 性能不高
   *
   * @param inputRddValue
   */
  def wordCount02(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResultRdd: RDD[(String, Int)] = rdd.map((_, 1))
    val groupByKeyRdd: RDD[(String, Iterable[Int])] = mapResultRdd.groupByKey
    val result: RDD[(String, Int)] = groupByKeyRdd.mapValues(item => item.size)
    result.collect.foreach(println)
  }


  /**
   * reduceByKey 性能高
   *
   * @param inputRddValue
   */
  def wordCount03(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResultRdd: RDD[(String, Int)] = rdd.map((_, 1))
    val result: RDD[(String, Int)] = mapResultRdd.reduceByKey(_ + _)
    result.collect.foreach(println)
  }

  /**
   * aggregateByKey 性能高
   *
   * @param inputRddValue
   */
  def wordCount04(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResultRdd: RDD[(String, Int)] = rdd.map((_, 1))
    val result: RDD[(String, Int)] = mapResultRdd.aggregateByKey(0)(_ + _, _ + _)
    result.collect.foreach(println)
  }


  /**
   * foldByKey
   *
   * @param inputRddValue
   */
  def wordCount05(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResultRdd: RDD[(String, Int)] = rdd.map((_, 1))
    val result: RDD[(String, Int)] = mapResultRdd.foldByKey(0)(_ + _)
    result.collect.foreach(println)
  }

  /**
   * combineByKey
   *
   * @param inputRddValue
   */
  def wordCount06(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResultRdd: RDD[(String, Int)] = rdd.map((_, 1))
    val result: RDD[(String, Int)] = mapResultRdd.combineByKey(
      v => v,
      (x: Int, y: Int) => x + y,
      (x: Int, y: Int) => x + y
    )
    result.collect.foreach(println)
  }

  /**
   * countByKey
   *
   * @param inputRddValue
   */
  def wordCount07(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResultRdd: RDD[(String, Int)] = rdd.map((_, 1))
    println(mapResultRdd.countByKey)
  }

  /**
   * countByKey
   *
   * @param inputRddValue
   */
  def wordCount08(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    println(rdd.countByValue)
  }


  /**
   * reduce,aggregate,fold都可以实现
   *
   * @param inputRddValue
   */
  def wordCount09(inputRddValue: RDD[String]): Unit = {
    val rdd: RDD[String] = inputRddValue.flatMap(_.split(" "))
    val mapResult: RDD[mutable.Map[String, Long]] = rdd.map(item => mutable.Map[String, Long]((item, 1)))
    val stringToLong: mutable.Map[String, Long] = mapResult.reduce(
      (map1, map2) => {
        map2.foreach {
          case (word, count) =>
            // 从map1中获取word对应的次数
            val newWordCountValue: Long = map1.getOrElse(word, 0L) + count
            map1.update(word, newWordCountValue)
        }
        map1
      }
    )
    println(stringToLong)
  }
}
